import org.apache.hadoop.hbase.spark.HBaseContext
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.Scan
import org.apache.hadoop.hbase.util.Bytes
import org.apache.log4j.Logger
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming._
import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods.parse
//
//报错:
//org.apache.hadoop.hbase.DoNotRetryIOException: java.lang.ExceptionInInitializerError
//at org.apache.hadoop.hbase.client.RpcRetryingCallerImpl.translateException(RpcRetryingCallerImpl.java:216)
//at org.apache.hadoop.hbase.client.RpcRetryingCallerImpl.callWithoutRetries(RpcRetryingCallerImpl.java:189)
//at org.apache.hadoop.hbase.client.ScannerCallableWithReplicas$RetryingRPC.call(ScannerCallableWithReplicas.java:393)
object Main {
  def main(args: Array[String]): Unit = {
    println("Hello world!")
    // 1. 创建SparkSession
    val spark = SparkSession.builder()
      .appName("HBaseRangeScanWithSpark")
      .master("local[*]") // 生产环境去掉这行，使用spark-submit参数
      .getOrCreate()

    val sc = spark.sparkContext

    // 2. 配置HBase
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "192.168.1.208:7077,192.168.1.208:7078,192.168.1.208:7079")
    //conf.set("hbase.zookeeper.quorum", "192.168.1.208")
    //conf.set("hbase.zookeeper.property.clientPort", "7077")
    conf.set("zookeeper.znode.parent", "/hbase")

    // 3. 创建HBaseContext
    val hbaseContext = new HBaseContext(sc, conf)

    // 4. 定义Scan对象
    val scan = new Scan()
    scan.withStartRow(Bytes.toBytes("0000000000000000")) // 起始行键
      .withStopRow(Bytes.toBytes("FFFFFFFFFFFFFFFF"))   // 结束行键
      .addColumn(Bytes.toBytes("route"), Bytes.toBytes("accessible")) // 指定列族和列
      .setLimit(10).setCaching(10) // 测试！限制返回十条数据，减少每次RPC返回的行数
      .setCacheBlocks(false) // 禁用块缓存

      //.setMaxResultSize(10485760) // 限制结果大小(10MB)

    // 5. 使用Spark读取HBase数据
    val hbaseRDD = hbaseContext.hbaseRDD(
      TableName.valueOf("mapdata"),
      scan
    )

    // 6. 处理数据
    val resultRDD = hbaseRDD.map { case(_, result) =>
      val rowKey = Bytes.toString(result.getRow)
      val arrayBytes = result.getValue(Bytes.toBytes("route"), Bytes.toBytes("accessible"))
      // arrayBytes类似这样b'[1,1,1,1,1,1,1,...,1,1,1]'，它不会以0结尾。
      val arrayJson = if (arrayBytes!=null) Bytes.toString(arrayBytes) else "[]"
      // 转成字符串"[1,1,1,1,1,1,1,...,1,1,1]"
      // 使用json4s解析JSON数组
      implicit val formats = DefaultFormats
      val array = parse(arrayJson).extract[List[Int]] // 根据实际数组元素类型调整
      (rowKey, array)
    }

    // 6. 打印结果
    println("===== 10 Records read from HBase =====")
    resultRDD.foreach { case (rowKey, array) =>
      println(s"RowKey: $rowKey")
      println(s"Array: ${array.mkString("[", ", ", "]")}")
      println("-----------------------------")
    }

    println("RDD数量是：", resultRDD.count())

//    // 7. 转换为DataFrame
//    import spark.implicits._
//    val df = resultRDD.toDF("rowkey", "value")
//
//    // 8. 显示结果
//    df.show(10, false)
//
//    // 9. 保存结果（可选）
//    df.write.parquet("hdfs://path/to/output")

    spark.stop()
  }
}